package org.kth.dks.old_tests;

import java.io.IOException;

import org.kth.dks.DKSAppInterface;
import org.kth.dks.DKSCallbackInterface;
import org.kth.dks.DKSImpl;
import org.kth.dks.DKSObject;
import org.kth.dks.dks_comm.ConnectionManager;
import org.kth.dks.dks_comm.DKSOverlayAddress;
import org.kth.dks.dks_comm.DKSRef;
import org.kth.dks.dks_dht.DKSDHTImpl;
import org.kth.dks.dks_exceptions.DKSTooManyRestartJoins;
import org.kth.dks.util.DKSPrintTypes;

public class SmallTest {

  long sendCounter = 0;
  long sendDroped = 0;

  class AppHandler
      implements DKSAppInterface {
    DKSCallbackInterface dks = null;
    public AppHandler() {}

    public void setDKSCallbacks(DKSCallbackInterface dksImpl) {
      dks = dksImpl;
    }

    public void joinCallback(DKSRef pred, DKSRef succ) {
      dks.joinCallbackReturn();
    }

    public void leaveCallback(DKSRef pred, DKSRef succ) {
      dks.leaveCallbackReturn();
    }

    public void broadcastCallback(DKSObject payload) {}

    public void sendStateCallback(DKSRef pred, DKSRef remoteid,
                                  DKSObject payload) {};

    public DKSObject routeCallback(long identifier, DKSObject payload) {
      String input = new String(payload.getData()).toUpperCase();
      return new DKSObject( (
          "test:tillbaka payload (callback mste overridas) indata:" + input).
                           getBytes());
    }

    public void routeCallbackAsync(long identifier, DKSObject payload) {
      System.out.println(
          "* Received a routeCallbackAsync() with data string : " +
          identifier + "=" + new String(payload.getData()));
    }

    public void failCallback(DKSRef failed, DKSRef failedPred) {

    }
  }

  public static void main(String[] args) {

    System.out.println("Starting DKS");
    if (args.length != 6 && args.length != 4 && args.length != 2) {
      System.out.println(
                       "Parameter: total nodes id1 firstnode firstnodeurl firstnodeport");
      System.out.println("Parameter: id1 url1 id2 url2");
      System.out.println("Parameter: id1 url1");
      System.exit(0);
    }
    SmallTest myTestBed = new SmallTest();
    //myTestBed.doTest(Integer.parseInt(args[0]), Integer.parseInt(args[1]));
    myTestBed.testDHTOnly();
//    myTestBed.testRoutingOnly();
    //myTestBed.doTest2(Integer.parseInt(args[0]), Integer.parseInt(args[1]));
  }

  void testDHTOnly() {
    ConnectionManager cm = null;
    try {
      cm = ConnectionManager.newInstance(4440); // Static factory only used for testing, should use getInstance (singleton)
    }
    catch (IOException io) {
      io.printStackTrace();
      return;
    }

    DKSDHTImpl dksobj = null;
    try {
      dksobj = new DKSDHTImpl(cm, 0, new java.net.URL("http://localhost:4440"));
      dksobj.logLevel(1);
      dksobj.join(0, null);
    }
    catch (Exception ex) {
      ex.printStackTrace();
      System.exit(0);
    }
    System.out.println("*** NODE 0 CREATED ***");

    int s = 10000;
    byte[] array = new byte[s];
    for (int x = 0; x < array.length; x++) {
      array[x] = (byte) x;
    }
    DKSObject a = new DKSObject(array);

    dksobj.addToBinding(50, a);

    ConnectionManager cm2 = null;
    try {
      cm2 = ConnectionManager.newInstance(4441); // Static factory only used for testing, should use getInstance (singleton)
    }
    catch (IOException ex) {
      ex.printStackTrace();
      return;
    }

    DKSDHTImpl dksobj2 = null;
    try {
      dksobj2 = new DKSDHTImpl(cm2, 50,
                               new java.net.URL("http://localhost:4441"));
      dksobj2.logLevel(0);

      boolean inside = false;
      while (!inside) {
        try {
          sleep(500);
          dksobj2.join(0, new java.net.URL("http://localhost:4440"));
          inside=true;
        }
        catch (DKSTooManyRestartJoins ex) {
          System.out.println("Too many restart joins");
        }
      }
    }
    catch (Exception ex) {
      StackTraceElement[] st = ex.getStackTrace();
      for (int y = 0; y < st.length; y++) {
        System.err.println(st[y].toString());
      }
    }

    DKSDHTImpl dksobj3 = null;
    try {
      dksobj3 = new DKSDHTImpl(cm2, 100,
                               new java.net.URL("http://localhost:4441"));
      dksobj3.logLevel(1);
      dksobj3.join(0, new java.net.URL("http://localhost:4440"));
    }
    catch (Exception ex) {
      ex.printStackTrace();
    }

    System.out.println("*** NODE 50 CREATED ***");

    dksobj.addToBinding(10, new DKSObject("hejsan svejsan".getBytes()));

    DKSObject[] arr = dksobj2.lookupBinding(10);
    System.out.println("Svar: " + new String(arr[0].getData()));

    sleep(1000);

    dksobj.leave();

    sleep(2500);

    dksobj2.leave();

    sleep(500);

    cm.end();

    cm2.end();
  }

  void testRoutingOnly() {
    int port = 4440;
    ConnectionManager cm = null;
    try {
      cm = ConnectionManager.newInstance(port); // Static factory only used for testing, should use getInstance (singleton)

      DKSImpl dksobj1 = new DKSImpl(cm, new DKSOverlayAddress(1022, 0, 0));
      String boot = dksobj1.getDKSURL();
      dksobj1.logLevel(0);
      dksobj1.create();

      System.out.println("*** NODE " + dksobj1.getDKSURL() + " CREATED ***");

      DKSImpl dksobj2 = new DKSImpl(cm, new DKSOverlayAddress(512, 0, 0));
      dksobj2.logLevel(0);
      dksobj2.join(DKSRef.valueOf(boot));

      DKSImpl dksobjJoker = new DKSImpl(cm, new DKSOverlayAddress(511, 0, 0));
      dksobjJoker.logLevel(0);
      dksobjJoker.join(DKSRef.valueOf(boot));

      DKSImpl dksobjJoker2 = new DKSImpl(cm, new DKSOverlayAddress(513, 0, 0));
      dksobjJoker2.logLevel(0);
      dksobjJoker2.join(DKSRef.valueOf(boot));

      System.err.println("Waiting before joining 0");
      sleep(1000);
      System.err.println("joining 0");
      sleep(1000);

      System.out.println("*** NODE " + dksobj2.getDKSURL() + " CREATED ***");
      sleep(200);
      System.out.println("\n##finaljoin\n");
      sleep(200);
//    ConnectionManager cm2 = ConnectionManager.newInstance(4441);

      DKSImpl dksobj3 = new DKSImpl(cm, new DKSOverlayAddress(1, 0, 0));
      dksobj3.logLevel(0);
      dksobj3.join(DKSRef.valueOf(boot));

      System.out.println("*** NODE " + dksobj3.getDKSURL() + " CREATED ***");
      DKSRef ref = dksobj2.findResponsible(51);
      System.out.println("*** LOOKED UP NODE 1 ITEM 0 *** RESP for 51 is " +
                         ref.getID());
      ref = dksobj2.findResponsible(20);
      System.out.println("*** LOOKED UP NODE 1 ITEM 0 *** RESP for 20 is " +
                         ref.getID());
      AppHandler appHandler = new AppHandler();
      DKSCallbackInterface dks = dksobj3.setCallbackHandler(appHandler);
      appHandler.setDKSCallbacks(dks);

      DKSObject obj = new DKSObject("input...".getBytes());
      sleep(500);
      dksobj1.routeAsync(99, obj);
//    DKSObject back = dksobj1.route(99, obj);
//    System.out.println("*** LOOKED UP NODE 0 ITEM 30 result: " +
//                       new String(back.getData()));
      sleep(1000);
      sleep(2000);
      dksobj3.leave();
      dksobj1.leave();
      dksobj2.leave();
      dksobjJoker.leave();
      dksobjJoker2.leave();
      sleep(2000);

//    cm2.end();
      cm.end();

      System.out.println("Done");
    }
    catch (Exception ex) {
      ex.printStackTrace();
    }

  }

  static void sleep(long msec) {
    try {
      Thread.sleep(msec);
    }
    catch (Exception ex) {
      ex.printStackTrace();
    }
  }

}
